package org.example.ItemCF.src.recommend;

import java.io.IOException;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.example.ItemCF.src.hdfs.HdfsDAO;

/**
 * 第一步：将用户-项目-偏好转换为用户-项目向量格式。
 * 此步骤主要包含两个部分：Mapper和Reducer。
 * Mapper负责将原始数据转换为用户-项目对的形式，
 * Reducer则将相同用户的项目偏好聚合在一起。
 */
public class Step1Driver {





    /**
     * 执行第一步的转换作业，主要完成从原始数据到用户-物品矩阵初步构建的处理过程。
     *
     * @param path 包含输入输出路径信息的映射表，其中应包括"Step1Input"和"Step1Output"两个键，
     *             分别对应第一步处理的输入路径和输出路径。
     * @throws IOException 当处理文件系统操作时发生错误，例如读取或写入文件时。
     * @throws ClassNotFoundException 当查找类时发生错误，可能在加载作业执行所需的类时出现。
     * @throws InterruptedException 当作业执行被中断时抛出，可能发生在作业等待完成的过程中。
     */
    public static void run(Map<String, String> path) throws IOException, ClassNotFoundException, InterruptedException {

        // 初始化Hadoop作业的配置和输入输出路径
        Configuration conf = new Configuration();
        String input = path.get("Step1Input");
        String output = path.get("Step1Output");

        // 使用HdfsDAO进行HDFS操作：清除旧输入路径、创建新输入路径、复制数据到输入路径
        HdfsDAO hdfs = new HdfsDAO(Recommend.HDFS, conf);
        hdfs.rmr(input); // 清除输入路径
        hdfs.mkdirs(input); // 创建输入路径
        hdfs.copyFile(path.get("data"), input); // 复制数据到输入路径

        // 设置作业的配置，包括作业名称、JAR包、Mapper和Reducer类以及输入输出格式
        Job job = Job.getInstance(conf, "RecommendStep1");
        job.setJarByClass(Step1Driver.class);
        job.setMapOutputKeyClass(IntWritable.class);
        job.setMapOutputValueClass(Text.class);

        job.setOutputKeyClass(IntWritable.class);
        job.setOutputValueClass(Text.class);

        job.setMapperClass(Step1ToItemPreMapper.class);
        job.setReducerClass(Step1ToUserVectorReducer.class);

        // 设置作业的输入输出路径
        FileInputFormat.addInputPath(job, new Path(input));
        FileOutputFormat.setOutputPath(job, new Path(output));

        // 提交作业并等待其完成
        if (!job.waitForCompletion(true))
            return; // 如果作业未成功完成，则返回

    }


}
